package net.sf.ehcache.store;

import com.google.common.base.Strings;
import com.je.core.service.MetaService;
import com.je.core.service.MetaServiceImpl;
import com.je.core.util.SpringContextHolder;
import net.sf.ehcache.CacheException;
import net.sf.ehcache.Ehcache;
import net.sf.ehcache.Element;
import net.sf.ehcache.distribution.RemoteCacheException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisSentinelPool;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;

/**
 * 二级缓存启动加载器
 *
 * @ProjectName: je-saas-platform
 * @Package: net.sf.ehcache.store
 * @ClassName: SecondLevelStoreBootstrapCacheLoader
 * @Description: 二级缓存启动加载器
 * @Author: LIULJ
 * @Version: 1.0
 * <p>Copyright: Copyright (c) 2018</p>
 */
public class SecondLevelStoreBootstrapCacheLoader extends MemoryLimitedCacheLoader {

    private static final Logger LOG = LoggerFactory.getLogger(SecondLevelStoreBootstrapCacheLoader.class);

    private static final String CACHE_TYPE_SINGLE = "single";
    private static final String CACHE_TYPE_CLUSTER = "cluster";
    private static final String CACHE_TYPE_SENTINEL = "sentinel";

    private final boolean asynchronous;
    private final long delay;
    private int sharding;
    private volatile boolean doneLoading;
    private volatile int loadedElements;

    /**
     * 延时几秒后开始在第0个db上加载
     *
     * @param delay
     */
    SecondLevelStoreBootstrapCacheLoader(final long delay) {
        asynchronous = true;
        this.delay = delay;
        this.sharding = 0;
    }

    /**
     * 延时几秒后在第sharding上加载
     *
     * @param delay
     * @param sharding
     */
    SecondLevelStoreBootstrapCacheLoader(final long delay, final int sharding) {
        asynchronous = true;
        this.delay = delay;
        this.sharding = sharding;
    }

    /**
     * 是否异步加载
     *
     * @param asynchronous
     */
    public SecondLevelStoreBootstrapCacheLoader(final boolean asynchronous) {
        this.asynchronous = asynchronous;
        delay = 0;
        this.sharding = 0;
    }

    /**
     * 如果是异步加载，在sharding上加载缓存
     *
     * @param asynchronous
     * @param sharding
     */
    public SecondLevelStoreBootstrapCacheLoader(final boolean asynchronous, final int sharding) {
        this.asynchronous = asynchronous;
        delay = 0;
        this.sharding = sharding;
    }

    @Override
    public void load(Ehcache cache) throws CacheException {
        if (cache.getCacheConfiguration().isUseCluster() && cache.getCacheConfiguration().isUseTwoLevelCache()) {
            if (asynchronous) {
                BootstrapThread thread = new BootstrapThread(cache);
                thread.start();
            } else {
                doLoad(cache);
            }
        } else {
            LOG.warn("Cache '" + cache.getName() + "' isn't second level cache, nothing to laod from!");
        }
    }

    @Override
    public boolean isAsynchronous() {
        return asynchronous;
    }

    private void doLoad(Ehcache cache) {
        if (cache.getCacheConfiguration().isUseOneLevelCache() == false && cache.getCacheConfiguration().isUseTwoLevelCache() == true) {
            //如果不启用一级缓存,只用二级缓存,那么就不在给本地缓存
        } else {
            loadedElements = 0;
            try {
                Properties redisConfigProp = PropertiesLoaderUtils.loadAllProperties("redis.properties");
                String issentinel = redisConfigProp.getProperty("redis.issentinel");
                String host = redisConfigProp.getProperty("redis.host");
                int port = Integer.parseInt(redisConfigProp.getProperty("redis.port"));
                String password = redisConfigProp.getProperty("redis.pass");

                Properties clusterProp = PropertiesLoaderUtils.loadAllProperties("cluster.properties");
                int sharding = Integer.valueOf(clusterProp.getProperty("cluster.sharding"));
                String cacheType = clusterProp.getProperty("cluster.cacheType");

                if (Strings.isNullOrEmpty(host) || port == 0) {
                    throw new RuntimeException("the redis second level cache loader require the redis host port can not be null");
                }

                Jedis jedis = null;
                if (CACHE_TYPE_SINGLE.equals(cacheType)) {
                    if (!Strings.isNullOrEmpty(issentinel) && issentinel.equals("1")) {
                        JedisSentinelPool jedisSentinelPool = SpringContextHolder.getBean(JedisSentinelPool.class);
                        jedis = jedisSentinelPool.getResource();
                    } else {
                        jedis = new Jedis(host, port);
                        jedis.auth(password);
                    }
                    jedis.select(sharding);
                } else if (CACHE_TYPE_CLUSTER.equals(cacheType)) {
                    LOG.error("the second level cache do not support this type {} please contact jeplus", cacheType);
                    throw new RuntimeException(String.format("the second level cache do not support this type {} please contact jeplus", cacheType));
                } else if (CACHE_TYPE_SENTINEL.equals(cacheType)) {
                    LOG.error("the second level cache do not support this type {} please contact jeplus", cacheType);
                    throw new RuntimeException(String.format("the second level cache do not support this type {} please contact jeplus", cacheType));
                } else {
                    throw new RuntimeException(String.format("unknown second level cache type %s", cacheType));
                }
                RedisSerializer<?> hashKeySerializer = new StringRedisSerializer();
                RedisSerializer<?> hashValueSerializer = new JdkSerializationRedisSerializer();
                Map<byte[], byte[]> resultMap = jedis.hgetAll(cache.getName().getBytes());
                LOG.info(String.format("load cache %s from second level cache,the number is %s", cache.getName(), resultMap.size()));
                for (Map.Entry<byte[], byte[]> each : resultMap.entrySet()) {
                    Element element = new Element(hashKeySerializer.deserialize(each.getKey()), hashValueSerializer.deserialize(each.getValue()));
                    cache.put(element);
                    ++loadedElements;
                }
            } catch (IOException e) {
                LOG.error("load redis.properties error", e);
            } finally {
                doneLoading = true;
            }
            LOG.debug("Loaded {} elements from disk into heap for cache {}", loadedElements, cache.getName());
        }
    }

    /**
     * 异步加载器
     */
    private final class BootstrapThread extends Thread {
        private Ehcache cache;

        public BootstrapThread(Ehcache cache) {
            super("Bootstrap Thread for cache " + cache.getName());
            this.cache = cache;
            setDaemon(true);
            setPriority(Thread.NORM_PRIORITY);
        }

        @Override
        public final void run() {
            try {
                sleep(delay);
                try {
                    doLoad(cache);
                } catch (RemoteCacheException e) {
                    LOG.warn("Error asynchronously performing bootstrap. The cause was: " + e.getMessage(), e);
                }
            } catch (InterruptedException e) {
                interrupted();
            } finally {
                cache = null;
            }
        }
    }

}
